package org.mq.mymq.broker;

import io.netty.util.internal.StringUtil;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.config.ConsumeQueueOffsetLoader;
import org.mq.mymq.broker.config.GlobalPropertiesLoader;
import org.mq.mymq.broker.config.MyMqTopicLoader;
import org.mq.mymq.broker.core.CommitLogAppendHandler;
import org.mq.mymq.broker.core.ConsumeQueueAppendHandler;
import org.mq.mymq.broker.core.ConsumeQueueConsumeHandler;
import org.mq.mymq.broker.model.MyMqTopicModel;
import org.mq.mymq.broker.netty.broker.BrokerServer;
import org.mq.mymq.broker.netty.nameserver.NameServerClient;
import org.mq.mymq.broker.slave.SlaveSyncService;
import org.mq.mymq.common.enums.BrokerClusterModeEnum;

import java.io.IOException;

/**
 * @ClassName BrokerStartUp
 * @Description MQ启动入口
 * @Author jiarong_ye
 * @Date 2025/6/15 19:42
 * @Version 1.0
 */public class BrokerStartUp {
     private static GlobalPropertiesLoader globalPropertiesLoader;
     private static MyMqTopicLoader myMqTopicLoader;
     private static CommitLogAppendHandler commitLogAppendHandler;
     private static ConsumeQueueOffsetLoader consumeQueueOffsetLoader;
     private static ConsumeQueueAppendHandler consumeQueueAppendHandler;
     private static ConsumeQueueConsumeHandler consumeQueueConsumeHandler;
     private static SlaveSyncService slaveSyncService;
    /**
     * 初始化配置
     */
    private static void initProperties() throws IOException {
        // 全局属性加载
        globalPropertiesLoader = new GlobalPropertiesLoader();
        // 主题配置信息加载
        myMqTopicLoader = new MyMqTopicLoader();
        // 消费者组消费队列位置记录信息加载
        consumeQueueOffsetLoader = new ConsumeQueueOffsetLoader();
        commitLogAppendHandler = new CommitLogAppendHandler();
        consumeQueueConsumeHandler = new ConsumeQueueConsumeHandler();
        consumeQueueAppendHandler = new ConsumeQueueAppendHandler();

        globalPropertiesLoader.loadProperties();
        myMqTopicLoader.loadProperties();
        // 开启异步定时刷新主题信息到磁盘
        myMqTopicLoader.startRefreshMyMqTopicInfoTash();
        consumeQueueOffsetLoader.loadProperties();
        // 开启异步定时刷新消费进度信息到磁盘
        consumeQueueOffsetLoader.startRefreshConsumerQueueOffsetTask();

        // 提前mmap映射提高吞吐量
        for (MyMqTopicModel myMqTopicModel : CommonCache.getMyMqTopicModelsMap().values()) {
            String topicName = myMqTopicModel.getTopic();
            commitLogAppendHandler.prepareMMapLoading(topicName);
            consumeQueueAppendHandler.prepareConsumeQueue(topicName);
        }
        CommonCache.setConsumeQueueConsumeHandler(consumeQueueConsumeHandler);
        CommonCache.setCommitLogAppendHandler(commitLogAppendHandler);
        CommonCache.setConsumeQueueAppendHandler(consumeQueueAppendHandler);
     }

    /**
     * 初始化与nameserver的长链接通道
     */
     private static void initNameServerChannel() {
         // 主从同步链路：从节点链接主节点，主节点收到数据，同步发送给从节点
         NameServerClient nameServerClient = CommonCache.getNameServerClient();
         nameServerClient.initConnection();
         nameServerClient.sendRegistryMsg();
         if (!BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(CommonCache.getGlobalProperties().getBrokerClusterMode()) ||
                 "master".equals(CommonCache.getGlobalProperties().getBrokerClusterRole())) {
             return;
         }
         String masterAddress = nameServerClient.queryBrokerMasterAddress();
         if (!StringUtil.isNullOrEmpty(masterAddress)) {
             // 尝试链接主节点
             slaveSyncService = new SlaveSyncService();
             if (slaveSyncService.connectMasterBrokerNode(masterAddress)) {
                 CommonCache.setSlaveSyncService(slaveSyncService);
                 slaveSyncService.sendStartSyncMsg();
             }
         }
     }

     private static void initReBalanceJob() {
         CommonCache.getConsumerInstancePool().startReBalanceJob();
     }

     private static void initBrokerServer() throws InterruptedException {
         BrokerServer brokerServer = new BrokerServer(CommonCache.getGlobalProperties().getBrokerPort());
         brokerServer.startServer();
     }

    public static void main(String[] args) throws IOException, InterruptedException {
        //加载配置 ，缓存对象的生成
        initProperties();
        initNameServerChannel();
        initReBalanceJob();
        // 同步调用，会阻塞监听端口
        initBrokerServer();
    }

//    private static void testMultiThread() {
//        //模拟初始化文件映射
//        String topic = "order_cancel_topic";
//        String userServiceConsumeGroup = "user_service_group";
//        String orderServiceConsumeGroup = "order_service_group";
//        new Thread(() -> {
//            while (true) {
//                byte[] content = consumeQueueConsumeHandler.consume(topic, userServiceConsumeGroup, 0);
//                if (content != null && content.length != 0) {
//                    System.out.println(userServiceConsumeGroup + ",消费内容:" + new String(content));
//                    consumeQueueConsumeHandler.ack(topic, userServiceConsumeGroup, 0);
//                } else {
//                    try {
//                        Thread.sleep(1000);
//                    } catch (InterruptedException e) {
//                        throw new RuntimeException(e);
//                    }
//                }
//            }
//        }).start();
//
//        new Thread(() -> {
//            while (true) {
//                byte[] content = consumeQueueConsumeHandler.consume(topic, orderServiceConsumeGroup, 0);
//                if (content != null) {
//                    System.out.println(orderServiceConsumeGroup + ",消费内容:" + new String(content));
//                    consumeQueueConsumeHandler.ack(topic, orderServiceConsumeGroup, 0);
//                } else {
//                    try {
//                        Thread.sleep(1000);
//                    } catch (InterruptedException e) {
//                        throw new RuntimeException(e);
//                    }
//                }
//            }
//        }).start();
//
//        AtomicInteger i = new AtomicInteger();
//        new Thread(() -> {
//            try {
//                Thread.sleep(5000);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
//            while (true) {
//                try {
//                    commitLogAppendHandler.appendMsg(topic, ("message_" + (i.getAndIncrement())).getBytes());
//                    TimeUnit.MILLISECONDS.sleep(100);
//                } catch (IOException e) {
//                    throw new RuntimeException(e);
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
//            }
//        }).start();
//
//        System.out.println("开始多线程消费验证");
//    }

//    public static void consumeQueue(String topic, String consumerGroup) {
//        for (int i = 0; i < 1001; i++) {
//            byte[] consume = consumeQueueConsumeHandler.consume(topic, consumerGroup, 0);
//            if (consume != null) {
//                consumeQueueConsumeHandler.ack(topic, consumerGroup, 0);
//            }
//            try {
//                TimeUnit.MILLISECONDS.sleep(10);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
//        }
//    }
//    public static void addMessageToQueue(String topic) throws IOException {
//        for (int i = 0; i < 1000; i++) {
//            try {
//                commitLogAppendHandler.appendMsg(topic, "1234567890".getBytes());
//            } catch (IOException e) {
//                throw new RuntimeException(e);
//            }
//        }
//    }
}
